package com.xuxueli.xxlrpc.remoting.invoker;

import com.xuxueli.xxlrpc.registry.Register;
import com.xuxueli.xxlrpc.registry.impl.LocalRegister;
import com.xuxueli.xxlrpc.remoting.net.params.XxlRpcFutureResponse;
import com.xuxueli.xxlrpc.remoting.net.params.XxlRpcResponse;
import com.xuxueli.xxlrpc.util.XxlRpcException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.*;

/**
 * @Author：YH
 * @Description：
 * @Date:Created in 2022/5/29 16:20
 */
public class XxlRpcInvokerFactory {

    private static final Logger logger = LoggerFactory.getLogger(XxlRpcInvokerFactory.class);

    // ---------------------- default instance ----------------------
    private static volatile XxlRpcInvokerFactory instance = new XxlRpcInvokerFactory(LocalRegister.class, null);

    public static XxlRpcInvokerFactory getInstance() {
        return instance;
    }

    // ---------------------- config ----------------------
    private Class<? extends Register> serviceRegistryClass;
    private Map<String, String> serviceRegistryParam;
    /**
     * @Description:抽象类
     **/
    private Register register;

    public Register getRegister() {
        return register;
    }

    public XxlRpcInvokerFactory(Class<? extends Register> serviceRegistryClass,
                                Map<String, String> getServiceRegistryParam) {
        this.serviceRegistryClass = serviceRegistryClass;
        this.serviceRegistryParam = getServiceRegistryParam;
    }

    public Class<? extends Register> getServiceRegistryClass() {
        return serviceRegistryClass;
    }

    public void setServiceRegistryClass(Class<? extends Register> serviceRegistryClass) {
        this.serviceRegistryClass = serviceRegistryClass;
    }

    public Map<String, String> getGetServiceRegistryParam() {
        return serviceRegistryParam;
    }

    public void setGetServiceRegistryParam(Map<String, String> getServiceRegistryParam) {
        this.serviceRegistryParam = getServiceRegistryParam;
    }

    public void stop() throws Exception {

    }

    public void start() throws Exception {
        logger.info("XxlRpcInvokerFactory.start");
        // start registry
        if (serviceRegistryClass != null) {
            register = serviceRegistryClass.newInstance();
            register.start(serviceRegistryParam);
        }
    }

    // ---------------------- future-response pool ----------------------

    // XxlRpcFutureResponseFactory

    private ConcurrentMap<String, XxlRpcFutureResponse> futureResponsePool = new ConcurrentHashMap<String,
            XxlRpcFutureResponse>();

    public void setInvokerFuture(String requestId, XxlRpcFutureResponse futureResponse) {
        futureResponsePool.put(requestId, futureResponse);
    }

    public void removeInvokerFuture(String requestId) {
        futureResponsePool.remove(requestId);
    }

    public void notifyInvokerFuture(String requestId, final XxlRpcResponse xxlRpcResponse) {
        // get
        final XxlRpcFutureResponse futureResponse = futureResponsePool.get(requestId);
        if (futureResponse == null) {
            return;
        }

        // notify
        if (futureResponse.getInvokeCallback() != null) {
            // callback type
            try {
                executeResponseCallback(new Runnable() {
                    @Override
                    public void run() {
                        if (xxlRpcResponse.getErrorMsg() != null) {
                            futureResponse.getInvokeCallback().onFailure(new XxlRpcException(xxlRpcResponse.getErrorMsg()));
                        } else {
                            futureResponse.getInvokeCallback().onSuccess(xxlRpcResponse.getResult());
                        }
                    }
                });
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        } else {

            // other nomal type
            futureResponse.setResponse(xxlRpcResponse);
        }

        // do remove
        futureResponsePool.remove(requestId);

    }


    // ---------------------- response callback ThreadPool ----------------------

    private ThreadPoolExecutor responseCallbackThreadPool = null;

    public void executeResponseCallback(Runnable runnable) {

        if (responseCallbackThreadPool == null) {
            synchronized (this) {
                if (responseCallbackThreadPool == null) {
                    responseCallbackThreadPool = new ThreadPoolExecutor(
                            10,
                            100,
                            60L,
                            TimeUnit.SECONDS,
                            new LinkedBlockingQueue<Runnable>(1000),
                            new ThreadFactory() {
                                @Override
                                public Thread newThread(Runnable r) {
                                    return new Thread(r,
                                            "xxl-rpc, XxlRpcInvokerFactory-responseCallbackThreadPool-" + r.hashCode());
                                }
                            },
                            new RejectedExecutionHandler() {
                                @Override
                                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                                    throw new XxlRpcException("xxl-rpc Invoke Callback Thread pool is EXHAUSTED!");
                                }
                            });        // default maxThreads 300, minThreads 60
                }
            }
        }
        responseCallbackThreadPool.execute(runnable);
    }

    public void stopCallbackThreadPool() {
        if (responseCallbackThreadPool != null) {
            responseCallbackThreadPool.shutdown();
        }
    }

}
